#pragma once

#include <memory>
#include <thread>
#include <cmath>
#include <chrono>
#include <cassert>
#include <queue>
#include <concurrentqueue.h>
#include <mutex>
#include "model/ClientNode.h"
#include "protocol/routeinfo.pb.h"
#include "protocol/transport.pb.h"
 
#include "base/Uri.h"
#include "base/Traits.h"
#include "base/Message.h"
#include "base/Serializer.h"
#include "base/Define.h"
#include "producer/client/RouteInfoSelector.h"
#include <shared_mutex>


namespace zeroplus::mq::producer
{
    class ProducerMediator;
    class MessagePublisher
    {
    private:
        std::map<std::string, std::shared_ptr<ClientNode>> clientTable_;
        // 读写锁
        std::map<std::string, routeinfo::TopicRouteInfo> topicRouteInfoTable_;
        std::map<std::string, moodycamel::ConcurrentQueue<std::pair<Header, Bytes>>> pendingMessageTable_;
        RouteInfoSelector selector_;
        static constexpr int maxRetries_ = 5;         // 最大重试次数
        static constexpr int initialBackoffMs_ = 100; // 初始退避时间(毫秒)
        std::shared_ptr<ProducerMediator> mediator_;
        std::shared_mutex rwMtx_;

    private:
        bool PublishInternal(const std::shared_ptr<ClientNode> client, const Header &header, const Bytes &body, int retryCount = 0);

        std::shared_ptr<ClientNode> GetChannelByTopic(std::string topic);

    public:
        MessagePublisher(std::shared_ptr<ProducerMediator> mediator);

        ~MessagePublisher();

        bool Publish(const Header &header, const Bytes &body);

        void InternalReceive(const std::string &topic, const routeinfo::TopicRouteInfo &topicRouteInfo);

        void setTopicRouteInfoTable(const std::string &topic, const routeinfo::TopicRouteInfo &topicRouteInfo)
        {
            std::unique_lock<std::shared_mutex> lock(rwMtx_);
            topicRouteInfoTable_[topic] = topicRouteInfo;
        }

        std::map<std::string, routeinfo::TopicRouteInfo> getTopicRouteInfoTable()
        {
            std::shared_lock<std::shared_mutex> lock(rwMtx_);
            return topicRouteInfoTable_;
        }
    };
}
